
#include <stdlib.h>
#include "../aprtools/APRFile.h"
#include "../utils/Hash.h"
#include "DataStore.h"

#include "acquire_config.h"

/** Constructor for the data store class (not the data store actor,
 * this happens when the data is committed later).
 * 
 * @param adb The actor database to use to add or read existing data
 * stores.
 * 
 * @param iname The name of the datastore to create or read.
 * 
 * @param iowner The parent actor of this datastore. This can be ANY
 * actor that created it (workinstance or user), so that the hierarchy
 * can be properly traversed later. */
DataStore::DataStore(ActorsDB *adb, const char *iname, const char *iowner) : Actor(adb)
{
   existing = 0;
   invalid = 0;
   if (adb->testActorExists(iname)) existing = 1;
   if (existing == 1)
      invalid = (newadb->isEquivalentPermission(iname, iowner) == 0);
   data_store_id = strdup(iname);
   setName(iname);
   if (existing)
      getType();
   else
      setType(ACTOR_TYPE_DATASTORE);
}

/** Destructor for the data store. */
DataStore::~DataStore()
{
   if (data_store_id) free(data_store_id);
   data_store_id = NULL;
}

/** Starts a new data store.
 * 
 * @param msgin Incoming broker message detailing what the data store
 * will contain.
 * 
 * @param keydb Key database to add new data store specific keys.
 * 
 * @param decryptingkeyname The name of the key capable of decrypting
 * the data store.
 * 
 * @param t_mp Memory pool to use for allocations in this function.
 * 
 * @return A data store state code giving a data store status or
 * error. */
DataStore::datastorestate DataStore::startNew(BrokerMessage &msgin, KeyDB *keydb, const char *decryptingkeyname, apr_pool_t *t_mp)
{
   if (invalid) return STORE_ACCESS_DENIED;
   if (existing) return ALREADY_EXISTS;
   const APRTable *headertable = msgin.getHeaderTable();
   const char *sizep_str = headertable->get("Payload-Of-Sizes");
   if ((sizep_str == NULL) || (strlen(sizep_str) == 0))
   {
      errprint("could not find sizes payload indicator");
      return ERROR_PARSING_SIZES;
   }
   const char *hashp_str = headertable->get("Payload-Of-Hashes");
   if ((hashp_str == NULL) || (strlen(hashp_str) == 0))
   {
      errprint("could not find plain hashes payload indicator");
      return ERROR_PARSING_HASHES;
   }
   int sizep_int = atoi(sizep_str);
   int hashp_int = atoi(hashp_str);
   const char *size_str = msgin.getPayload(sizep_int);
   if ((size_str == NULL) || (strlen(size_str) == 0))
   {
      errprint("could not find sizes payload");
      return ERROR_PARSING_SIZES;
   }
   const char *hashes_str = msgin.getPayload(hashp_int);
   if ((hashes_str == NULL) || (strlen(hashes_str) == 0))
   {
      errprint("could not find hashes payload");
      return ERROR_PARSING_HASHES;
   }
   char *ult_user = newadb->getUltimateUser(msgin.getSender());
   if (ult_user == NULL)
   {
      errprint("could not find ultimate user of parent");
      return ERROR_FINDING_USER;
   }
   
   Actor user_actor(newadb);
   user_actor.setName(ult_user);
   user_actor.getType();
   int64_t quota_size = user_actor.getUserQuotaAsInt64();
   int64_t total_data_size = StringUtils::sumNumberList(size_str, ",") + strlen(size_str) + strlen(hashes_str);
   if (quota_size < total_data_size)
   {
      errprint("not enough quota: '%s' has quota %" APR_INT64_T_FMT ", data would be %" APR_INT64_T_FMT, ult_user, quota_size, total_data_size);
      return NOT_ENOUGH_QUOTA;
   }
   const char *unenc_block_size = headertable->get("Unencrypted-Block-Size");
   if ((unenc_block_size == NULL) || (strlen(unenc_block_size) == 0) || (atoi(unenc_block_size) <= 0))
   {
      errprint("unencrypted block size not found/empty/too small");
      return INVALID_BLOCK_SIZE_SPECIFIED;
   }
   datastorestate retres = createNewStoreFiles(size_str, hashes_str, ult_user, t_mp);
   std::vector<char *> *strlist = StringUtils::unflattenStringList(size_str, ",");
   if (retres == FILE_CREATION_SUCCEEDED)
      retres = createNewDataStoreActor(msgin, msgin.getProcessee(), atoi(unenc_block_size), strlist->size(), t_mp, keydb, decryptingkeyname);
   if ((retres != FILE_CREATION_SUCCEEDED) && (retres != ACTOR_CREATION_SUCCEEDED))
   {
      rollbackNewStoreFiles(size_str, hashes_str, ult_user, t_mp);
      rollbackDataStoreActor(data_store_id, keydb);
   }
   free(ult_user);
   StringUtils::destroyStringList(strlist);
   return retres;
}

/** Creates new dummy files for the datastore. This reduces the chance
 * that things break once the datastore has been accepted.
 * 
 * @param sizes_str A comma-separated string of sizes for each of the
 * blocks.
 * 
 * @param hashes_str A comma-separated string of hashes for each block.
 * 
 * @param ult_user The ultimate user for obtaining resources from.
 * 
 * @param t_mp Memory pool to use for allocations in this function.
 * 
 * @return A data store state code giving a data store status or
 * error. */
DataStore::datastorestate DataStore::createNewStoreFiles(const char *sizes_str, const char *hashes_str, const char *ult_user, apr_pool_t *t_mp)
{
   char *hashes_filename = StringUtils::varprintf(ACQUIRE_DYNAMIC_DIR "/datastore_%s.hashes", data_store_id);
   if (StringUtils::writeStringToFile(hashes_str, hashes_filename, t_mp))
   {
      errprint("could not write data to '%s'", hashes_filename);
      free(hashes_filename);
      return HASHES_FILE_ERROR;
   }
   free(hashes_filename);
   char *sizes_filename = StringUtils::varprintf(ACQUIRE_DYNAMIC_DIR "/datastore_%s.sizes", data_store_id);
   if (StringUtils::writeStringToFile(sizes_str, sizes_filename, t_mp))
   {
      errprint("could not write data to '%s'", sizes_filename);
      free(sizes_filename);
      return SIZES_FILE_ERROR;
   }
   free(sizes_filename);
   int blockno = 0;
   int no_of_blocks = 0;
   std::vector<char *> *sizes_list = StringUtils::unflattenStringList(sizes_str, ",");
   if (sizes_list)
   {
      int rollback = 0;
      no_of_blocks = sizes_list->size();
      for (int i = 0; i < no_of_blocks; i++)
      {
         if ((sizes_list->at(i)) && (atoi(sizes_list->at(i)) > 0))
         {
            char *blockdata_filename = StringUtils::varprintf(ACQUIRE_DYNAMIC_DIR "/datastore_%s.iblock%d", data_store_id, i);
            APRFile::blockOutFileSpace(blockdata_filename, atoi(sizes_list->at(i)), &rollback, t_mp);
            free(blockdata_filename);
         }
      }
      StringUtils::destroyStringList(sizes_list);
      if (rollback) return FILE_CREATION_FAILED;
   } else
   {
      errprint("error finding sizes list?");
      return ERROR_PARSING_SIZES;
   }
   return FILE_CREATION_SUCCEEDED;
}

/** Creates the actor (database resident) part of the datastore.
 *
 * @param msgin The message that was sent, containing information on
 * how to build the datastore actor.
 *
 * @param ult_user The ultimate user with usable resources that is the
 * root of the chain.
 *
 * @param tblocksize The plain data block size to upload.
 *
 * @param no_of_blocks The number of blocks to upload.
 *
 * @param t_mp The memory pool to use.
 *
 * @param nkdb The key database.
 *
 * @param decryptingkeyname The name of the decryption key.
 *
 * @return The final state of the datastore, or an error. */
DataStore::datastorestate DataStore::createNewDataStoreActor(BrokerMessage &msgin, const char *ult_user,
   int tblocksize, int no_of_blocks, apr_pool_t *t_mp, KeyDB *nkdb, const char *decryptingkeyname)
{
   SQLiteAutoLock thislock(newadb->getAutoLock());
   // Need to actually create the data store Actor now
   newadb->addDataStore(data_store_id, ult_user, ult_user, NULL, decryptingkeyname, no_of_blocks, tblocksize);
   // Need to now add their keys to the system
   msgin.obtainAllKeysFromReceived(0, "", "", ult_user, false, data_store_id, nkdb);
   const APRTable *headertable = msgin.getHeaderTable();
   if (headertable->get("WorkPacket-Decryption-Key-Slot"))
   {
      int payloadno = atoi(headertable->get("WorkPacket-Decryption-Key-Slot"));
      char *id_inp = StringUtils::varprintf("%s_inp", data_store_id);
      char *keyname_header = StringUtils::varprintf("keyslot%d_deckeyname", payloadno);
      // if they are not equal, someone may come looking for the wrong key...
      if (strcmp(headertable->get(keyname_header), id_inp) != 0)
         newadb->addEquivalenceOnKey(id_inp, headertable->get(keyname_header));
      free(keyname_header);
      // now store it under the new name
      msgin.obtainKeyFromReceived(payloadno, "", "", Key::KEYPURPOSE_DECRYPTION, ult_user, false, id_inp, nkdb);
      newadb->addDependencyOnKey(data_store_id, id_inp);
      free(id_inp);
   }
   if (headertable->get("Results-Encryption-Key-Slot"))
   {
      int payloadno = atoi(headertable->get("Results-Encryption-Key-Slot"));
      char *id_out = StringUtils::varprintf("%s_out", data_store_id);
      char *keyname_header = StringUtils::varprintf("keyslot%d_enckeyname", payloadno);
      // if they are not equal, someone may come looking for the wrong key...
      if (strcmp(headertable->get(keyname_header), id_out) != 0)
         newadb->addEquivalenceOnKey(id_out, headertable->get(keyname_header));
      free(keyname_header);
      // now store it under the new name
      msgin.obtainKeyFromReceived(payloadno, "", "", Key::KEYPURPOSE_ENCRYPTION, ult_user, false, id_out, nkdb);
      newadb->addDependencyOnKey(data_store_id, id_out);
      free(id_out);
   }
   // only necessary for chaining workpackets - so isn't the critical path.
   // key names have already been added to the respective databases by the other ifs
   if (headertable->get("WorkPacket-Encryption-Key-Slot"))
   {
      int payloadno = atoi(headertable->get("WorkPacket-Encryption-Key-Slot"));
      char *id_inp = StringUtils::varprintf("%s_inp", data_store_id);
      msgin.obtainKeyFromReceived(payloadno, "", "", Key::KEYPURPOSE_ENCRYPTION, ult_user, false, id_inp, nkdb);
      free(id_inp);
   }
   newadb->addDependencyOnKey(data_store_id, data_store_id);
   return ACTOR_CREATION_SUCCEEDED;
}

/** Roll back the data store actor. This is done when some part of the
 * allocation and start-up process fails.
 *
 * @param actorname The name of the data store actor.
 *
 * @param keydb The key database (since the actor probably has
 * dependent keys). */
void DataStore::rollbackDataStoreActor(const char *actorname, KeyDB *keydb)
{
   newadb->removeActorAndKeys(actorname, keydb);
}

/** Roll back the allocation and more physical aspects of the data
 * store.
 *
 * @param size_str The comma-separated sizes of the compressed,
 * encrypted blocks.
 *
 * @param hashes_str The comma-separated hashes of the plain text
 * blocks.
 *
 * @param ult_user The ultimate user account that owns
 * everything. Needed to restore the freed resources to them.
 *
 * @param t_mp The memory pool to use for allocating file handles. */
void DataStore::rollbackNewStoreFiles(const char *size_str, const char *hashes_str, const char *ult_user, apr_pool_t *t_mp)
{
   errprint("data store creation error, rollback initiated");
   // do the roll back of the data
   char *hashes_filename = StringUtils::varprintf(ACQUIRE_DYNAMIC_DIR "/datastore_%s.hashes", data_store_id);
   APRFile::removeWrapper(hashes_filename, t_mp);
   free(hashes_filename);
   char *sizes_filename = StringUtils::varprintf(ACQUIRE_DYNAMIC_DIR "/datastore_%s.sizes", data_store_id);
   APRFile::removeWrapper(sizes_filename, t_mp);
   free(sizes_filename);
   std::vector<char *> *sizes_list = StringUtils::unflattenStringList(size_str, ",");
   if (sizes_list)
   {
      for (int i = 0; i < sizes_list->size(); i++)
      {
         char *blockdata_filename = StringUtils::varprintf(ACQUIRE_DYNAMIC_DIR "/datastore_%s.iblock%d", data_store_id, i);
         APRFile::removeWrapper(blockdata_filename, t_mp);
         free(blockdata_filename);
      }
   }
   StringUtils::destroyStringList(sizes_list);
}

/** Check the status of the given block.
 *
 * @param headertable The header table derived from the input message.
 *
 * @param t_mp The memory pool to use.
 *
 * @return The data store block status or an error. */
DataStore::datastorestate DataStore::checkBlockStatus(const APRTable *headertable, apr_pool_t *t_mp)
{
   if (invalid) return BLOCK_ACCESS_DENIED;
   const char *block_number_string = headertable->get("Block-ID");
   const char *block_plain_hash = headertable->get("Block-Plain-Hash");
   const char *block_size = headertable->get("Block-Size");
   if ((block_number_string == NULL) || (strlen(block_number_string) == 0))
   {
      errprint("could not find block number");
      return MESSAGE_CORRUPT;
   }
   if ((block_plain_hash == NULL) || (strlen(block_plain_hash) == 0))
   {
      errprint("could not find block plain hash");
      return MESSAGE_CORRUPT;
   }
   if ((block_size == NULL) || (strlen(block_size) == 0))
   {
      errprint("could not find block size");
      return MESSAGE_CORRUPT;
   }
   int comp_data_blockid;
   sscanf(block_number_string, "%d", &comp_data_blockid);
   if ((comp_data_blockid < 0) || (comp_data_blockid >= getDataStoreBlocksHeldAsInt()))
   {
      errprint("block id is invalid (%d must be >= 0, < %d) in datastore '%s'", comp_data_blockid, getDataStoreBlocksHeldAsInt(), data_store_id);
      return MESSAGE_CORRUPT;
   }
   apr_size_t comp_data_size;
   sscanf(block_size, "%" APR_INT64_T_FMT, &comp_data_size);
   return checkFileBlockStatus(comp_data_blockid, block_plain_hash, comp_data_size, t_mp);
}

/** Check the status of the given file block.
 *
 * @param blockid The number of the block to check.
 *
 * @param plainhash The plain text hash of the data, to compare to the
 * hash that has been promised.
 *
 * @param datasize The size of the block when compressed and
 * encrypted.
 *
 * @param t_mp The pool to allocate from.
 *
 * @return A status code for the current block status. */
DataStore::datastorestate DataStore::checkFileBlockStatus(const int blockid, const char *plainhash, const apr_size_t datasize, apr_pool_t *t_mp)
{
   int retread = 0;
   if (invalid) return BLOCK_ACCESS_DENIED;
   char *hashes_filename = StringUtils::varprintf(ACQUIRE_DYNAMIC_DIR "/datastore_%s.hashes", data_store_id);
   char *block_string = NULL;
   retread = StringUtils::readFileToString(&block_string, hashes_filename, t_mp);
   if ((retread != 0) || (block_string == NULL))
   {
      errprint("error reading '%s'", hashes_filename);
      free(hashes_filename);
      return HASHES_FILE_ERROR;
   }
   free(hashes_filename);
   std::vector<char *> *all_hashes = StringUtils::unflattenStringList(block_string, ",");
   free(block_string);
   if ((blockid < 0) || (blockid >= (all_hashes->size())))
   {
      StringUtils::destroyStringList(all_hashes);
      return INVALID_PLAINTEXT_HASH_SPECIFIED;
   }
   if (strcmp(plainhash, all_hashes->at(blockid)) != 0)
   {
      errprint("block hash mismatch %s != %s", plainhash, all_hashes->at(blockid));
      StringUtils::destroyStringList(all_hashes);
      return WRONG_PLAINTEXT_HASH;
   }
   StringUtils::destroyStringList(all_hashes);
   char *sizes_filename = StringUtils::varprintf(ACQUIRE_DYNAMIC_DIR "/datastore_%s.sizes", data_store_id);
   char *sizes_string = NULL;
   retread = StringUtils::readFileToString(&sizes_string, sizes_filename, t_mp);
   if ((retread != 0) || (sizes_string == NULL))
   {
      errprint("error reading '%s'", sizes_filename);
      free(sizes_filename);
      return SIZES_FILE_ERROR;
   }
   free(sizes_filename);
   std::vector<char *> *all_sizes = StringUtils::unflattenStringList(sizes_string, ",");
   free(sizes_string);
   if ((blockid < 0) || (blockid >= (all_sizes->size())))
   {
      StringUtils::destroyStringList(all_sizes);
      return INVALID_BLOCK_SIZE_SPECIFIED;
   }
   if ((all_sizes->at(blockid) == NULL) || (datasize != atoi(all_sizes->at(blockid))))
   {
      errprint("block size mismatch %" APR_SIZE_T_FMT " != %d", datasize, atoi(all_sizes->at(blockid)));
      StringUtils::destroyStringList(all_sizes);
      return MISMATCHED_BLOCK_SIZE;
   }
   StringUtils::destroyStringList(all_sizes);
   char *blockdata_c_filename = StringUtils::varprintf(ACQUIRE_DYNAMIC_DIR "/datastore_%s.cblock%d", data_store_id, blockid);
   char *blockdata_i_filename = StringUtils::varprintf(ACQUIRE_DYNAMIC_DIR "/datastore_%s.iblock%d", data_store_id, blockid);
   apr_size_t c_data_size = APRFile::getFileSize(blockdata_c_filename, t_mp);
   apr_size_t i_data_size = APRFile::getFileSize(blockdata_i_filename, t_mp);
   free(blockdata_c_filename);
   free(blockdata_i_filename);
   if ((c_data_size != datasize) && (i_data_size != datasize))
   {
      errprint("couldn't find allocated block");
      return NO_ALLOCATED_SPACE;
   }
   if (i_data_size == datasize) return DATA_BLOCK_INCOMPLETE;
   if (c_data_size == datasize) return DATA_BLOCK_COMPLETE;
   errprint("unknown error in data block");
   return MESSAGE_CORRUPT;
}

/** Write out a status broker message on the subject of this data
 * store.
 *
 * @param msgout The output message into which the status data will be
 * written.
 *
 * @param t_mp The memory pool to allocate from. */
void DataStore::writeOutStatus(BrokerMessage &msgout, apr_pool_t *t_mp)
{
   if ((!(invalid)) && (existing))
   {
      msgout.addTypeHeaderNOCRLF("DATA_STORE_STATUS",
         (checkActorMark(ActorsDB::ACTORMARK_DATA_COMPLETE)) ? "DATA_STORE_COMPLETE" : "DATA_STORE_INCOMPLETE");
      msgout.addHeaderLineNOCRLF("DS-WorkPacket: %s", (checkActorMark(ActorsDB::ACTORMARK_DATA_IS_WORK)) ? "true" : "false");
      msgout.addHeaderLineNOCRLF("DS-Results: %s", (checkActorMark(ActorsDB::ACTORMARK_DATA_IS_RESULTS)) ? "true" : "false");
      msgout.addHeaderLineNOCRLF("DS-Blocks: %d", getDataStoreBlocksHeldAsInt());
      msgout.addHeaderLineNOCRLF("DS-BlockSize: %" APR_INT64_T_FMT, getDataStoreBlockSizeAsInt64());
      msgout.addHeaderLineNOCRLF("DS-Origin: %s", getDataStoreOrigin());
      msgout.addHeaderLineNOCRLF("DS-Instances: %d", getDataStoreLocalIDNosAsInt());
      msgout.addHeaderLineNOCRLF("DS-Decryption-Key: %s", getDataStoreDecryptionKey());
      char *hashes_filename = StringUtils::mpprintf(t_mp, ACQUIRE_DYNAMIC_DIR "/datastore_%s.hashes", data_store_id);
      char *block_string = NULL;
      int retread = -1;
      retread = StringUtils::readFileToString(&block_string, hashes_filename, t_mp);
      if ((retread == 0) && (block_string != NULL))
      {
         int payload_no = msgout.addVerbatimPayloadString(block_string);
         msgout.addHeaderLineNOCRLF("DS-Payload-For-Hashes: %d", payload_no);
      }
      if (block_string) free(block_string);
      char *sizes_filename = StringUtils::mpprintf(t_mp, ACQUIRE_DYNAMIC_DIR "/datastore_%s.sizes", data_store_id);
      char *sizes_string = NULL;
      retread = StringUtils::readFileToString(&sizes_string, sizes_filename, t_mp);
      if ((retread == 0) && (sizes_string != NULL))
      {
         int payload_no = msgout.addVerbatimPayloadString(sizes_string);
         msgout.addHeaderLineNOCRLF("DS-Payload-For-Sizes: %d", payload_no);
         free(sizes_string);
         sizes_string = NULL;
      }
      if (sizes_string) free(sizes_string);
   } else
      msgout.addTypeHeaderNOCRLF("DATA_STORE_STATUS", "DATA_STORE_INVALID");
}

/** Write out a block of data to the disk.
 *
 * @param msgin The incoming message containing the data block to
 * write out.
 *
 * @param t_mp The memory pool to allocate from.
 *
 * @return The data store status, or the status of the block writing
 * process. */
DataStore::datastorestate DataStore::writeBlock(BrokerMessage &msgin, apr_pool_t *t_mp)
{
   if (invalid) return STORE_ACCESS_DENIED;
   const APRTable *headertable = msgin.getHeaderTable();
   const char *block_real_hash = headertable->get("Block-Real-Hash");
   if (block_real_hash == NULL) return TRANSFER_HASH_MISSING;
   const char *block_data_payload = headertable->get("Block-Data-Payload");
   if (block_data_payload == NULL) return TRANSFER_PAYLOAD_NUMBER_NOT_FOUND;
   int payloadno = atoi(block_data_payload);
   const char *data_buffer = msgin.getPayload(payloadno);
   if (data_buffer == NULL)
   {
      errprint("payload supposedly '%s'", block_data_payload);
      return TRANSFER_PAYLOAD_NOT_FOUND;
   }
   apr_size_t comp_data_size;
   sscanf(headertable->get("Block-Size"), "%" APR_INT64_T_FMT, &comp_data_size);
   if (msgin.getPayloadLength(payloadno) != comp_data_size)
   {
      errprint("payload length %d is not what was promised (%d)", (int)(msgin.getPayloadLength(payloadno)), (int)comp_data_size);
      return MISMATCHED_BLOCK_SIZE;
   }
   char *hex_hash = Hash::getHexSHA1Hash(data_buffer, comp_data_size, NULL);
   if (strcmp(hex_hash, headertable->get("Block-Real-Hash")) != 0)
   {
      errprint("hash of data expected to be '%s', was actually '%s'", headertable->get("Block-Real-Hash"), hex_hash);
      free(hex_hash);
      return TRANSFER_HASH_MISMATCH;
   }
   free(hex_hash);
   if (headertable->get("Block-ID") == NULL) return MESSAGE_CORRUPT;
   char *blockdata_c_filename = StringUtils::varprintf(ACQUIRE_DYNAMIC_DIR "/datastore_%s.cblock%d",
      msgin.getProcessee(), atoi(headertable->get("Block-ID")));
   char *blockdata_i_filename = StringUtils::varprintf(ACQUIRE_DYNAMIC_DIR "/datastore_%s.iblock%d",
      msgin.getProcessee(), atoi(headertable->get("Block-ID")));
   int ret = StringUtils::overwriteFileWithBuffer(data_buffer, comp_data_size, blockdata_i_filename, t_mp);
   if (ret != 0)
   {
      free(blockdata_i_filename);
      free(blockdata_c_filename);
      return DATA_FILE_ERROR;
   }
   ret = APRFile::renameWrapper(blockdata_i_filename, blockdata_c_filename, t_mp);
   free(blockdata_i_filename);
   free(blockdata_c_filename);
   if (ret != 0) return DATA_FILE_UNMOVED;
   return TRANSFER_SUCCEEDED;
}

/** Read out a block of data onto a message.
 *
 * @param msgout The outgoing message that is to contain the data
 * block.
 *
 * @param blockid The number of the block to read out onto the
 * message.
 *
 * @param t_mp The memory pool to allocate from.
 *
 * @return The data store status, or the status of the block writing
 * process. */
DataStore::datastorestate DataStore::readBlock(BrokerMessage &msgout, int blockid, apr_pool_t *t_mp)
{
   char *blockdata_c_filename = StringUtils::mpprintf(t_mp, ACQUIRE_DYNAMIC_DIR "/datastore_%s.cblock%d",
      getName(), blockid);
   char *blockdata_hash_filename = StringUtils::mpprintf(t_mp, ACQUIRE_DYNAMIC_DIR "/datastore_%s.hashes",
      getName(), blockid);
   char *blockdata_size_filename = StringUtils::mpprintf(t_mp, ACQUIRE_DYNAMIC_DIR "/datastore_%s.sizes",
      getName(), blockid);
   char *hash_data_string = NULL;
   int retval_hash = StringUtils::readFileToString(&hash_data_string, blockdata_hash_filename, t_mp);
   if ((retval_hash != 0) || (hash_data_string == NULL))
   {
      errprint("failed to get plain hashes");
      return HASHES_FILE_ERROR;
   }
   std::vector<char *> *hashes_vector = StringUtils::unflattenStringList(hash_data_string, ",");
   free(hash_data_string);
   const char *hash_for_block = StringUtils::mpprintf(t_mp, "%s", hashes_vector->at(blockid));
   StringUtils::destroyStringList(hashes_vector);
   char *size_data_string = NULL;
   int retval_sizes = StringUtils::readFileToString(&size_data_string, blockdata_size_filename, t_mp);
   if ((retval_sizes != 0) || (size_data_string == NULL))
   {
      errprint("failed to get sizes");
      return SIZES_FILE_ERROR;
   }
   std::vector<char *> *sizes_vector = StringUtils::unflattenStringList(size_data_string, ",");
   free(size_data_string);
   if ((sizes_vector->at(blockid) == NULL) || (strlen(sizes_vector->at(blockid)) == 0))
   {
      errprint("failed to get block %d size", blockid);
      return SIZES_FILE_ERROR;
   }
   // need to get hash and size data and verify that it is correct!
   char *file_block = NULL;
   int file_buffer_length = 0;
   int retval = StringUtils::readFileToBuffer(&file_block, &file_buffer_length, blockdata_c_filename, t_mp);
   int max_blocks = getDataStoreBlocksHeldAsInt();
   if ((blockid < 0) || (blockid >= max_blocks))
   {
      msgout.addTypeHeaderNOCRLF("DOWNLOAD_DATA_ERROR", "INVALID_BLOCK");
      errprint("invalid block number for '%s', block %d doesn't exist", getName(), blockid);
      StringUtils::destroyStringList(sizes_vector);
      if (file_block) free(file_block);
      return BLOCK_ACCESS_DENIED;
   }
   if ((file_block == NULL) || (retval != 0) || (file_buffer_length == 0))
   {
      msgout.addTypeHeaderNOCRLF("DOWNLOAD_DATA_ERROR", "DATA_FILE_ERROR");
      errprint("error with file '%s' while trying to serve block %d", blockdata_c_filename, blockid);
      StringUtils::destroyStringList(sizes_vector);
      return DATA_FILE_ERROR;
   }
   if (atoi(sizes_vector->at(blockid)) == 0)
   {
      errprint("%d size emitted for block %d", atoi(sizes_vector->at(blockid)), blockid);
      StringUtils::destroyStringList(sizes_vector);
      free(file_block);
      return MISMATCHED_BLOCK_SIZE;
   }
   msgout.addTypeHeaderNOCRLF("DOWNLOAD_DATA_BLOCK", "SUCCESS");
   msgout.addHeaderLineNOCRLF("Block-Plain-Hash: %s", hash_for_block);
   msgout.addHeaderLineNOCRLF("Block-ID: %d", blockid);
   int payloadno = msgout.addPayloadBuffer(file_block, atoi(sizes_vector->at(blockid)));
   msgout.addHeaderLineNOCRLF("Block-At-Payload: %d", payloadno);
   char *hex_hash = Hash::getHexSHA1Hash(file_block, file_buffer_length, NULL);
   msgout.addHeaderLineNOCRLF("Block-Real-Hash: %s", hex_hash);
   free(hex_hash);
   free(file_block);
   StringUtils::destroyStringList(sizes_vector);
   return TRANSFER_SUCCEEDED;
}

/** Turn the data store status, block status, process status or error
 * into a status string for appending to the subject part of an
 * outgoing message.
 *
 * @param tstate The state to turn into a subject string.
 *
 * @return The subject string. */
const char *DataStore::stateToMessageTag(DataStore::datastorestate tstate)
{
   const char *temp_msg = "";
   switch (tstate)
   {
      case STORE_CREATED: temp_msg = "STORE_CREATED"; break;
      case ALREADY_EXISTS: temp_msg = "ALREADY_EXISTS"; break;
      case STORE_ACCESS_DENIED: temp_msg = "STORE_ACCESS_DENIED"; break;
      case ERROR_PARSING_SIZES: temp_msg = "ERROR_PARSING_SIZES"; break;
      case ERROR_PARSING_HASHES: temp_msg = "ERROR_PARSING_HASHES"; break;
      case ERROR_FINDING_USER: temp_msg = "ERROR_FINDING_USER"; break;
      case NOT_ENOUGH_QUOTA: temp_msg = "NOT_ENOUGH_QUOTA"; break;
      case DATA_BLOCK_COMPLETE: temp_msg = "DATA_BLOCK_COMPLETE"; break;
      case DATA_BLOCK_INCOMPLETE: temp_msg = "DATA_BLOCK_INCOMPLETE"; break;
      case INVALID_PLAINTEXT_HASH_SPECIFIED: temp_msg = "INVALID_PLAINTEXT_HASH_SPECIFIED"; break;
      case WRONG_PLAINTEXT_HASH: temp_msg = "WRONG_PLAINTEXT_HASH"; break;
      case INVALID_BLOCK_SIZE_SPECIFIED: temp_msg = "INVALID_BLOCK_SIZE_SPECIFIED"; break;
      case MISMATCHED_BLOCK_SIZE: temp_msg = "MISMATCHED_BLOCK_SIZE"; break;
      case NO_ALLOCATED_SPACE: temp_msg = "NO_ALLOCATED_SPACE"; break;
      case MESSAGE_CORRUPT: temp_msg = "MESSAGE_CORRUPT"; break;
      case HASHES_FILE_ERROR: temp_msg = "HASHES_FILE_ERROR"; break;
      case SIZES_FILE_ERROR: temp_msg = "SIZES_FILE_ERROR"; break;
      case DATA_FILE_ERROR: temp_msg = "DATA_FILE_ERROR"; break;
      case BLOCK_ACCESS_DENIED: temp_msg = "BLOCK_ACCESS_DENIED"; break;
      case TRANSFER_HASH_MISSING: temp_msg = "TRANSFER_HASH_MISSING"; break;
      case TRANSFER_HASH_MISMATCH: temp_msg = "TRANSFER_HASH_MISMATCH"; break;
      case TRANSFER_PAYLOAD_NUMBER_NOT_FOUND: temp_msg = "TRANSFER_PAYLOAD_NUMBER_NOT_FOUND"; break;
      case TRANSFER_PAYLOAD_NOT_FOUND: temp_msg = "TRANSFER_PAYLOAD_NOT_FOUND"; break;
      case DATA_FILE_UNMOVED: temp_msg = "DATA_FILE_UNMOVED"; break;
      case FILE_CREATION_FAILED: temp_msg = "FILE_CREATION_FAILED"; break;
      case FILE_CREATION_SUCCEEDED: temp_msg = "FILE_CREATION_SUCCEEDED"; break;
      case ACTOR_CREATION_SUCCEEDED: temp_msg = "ACTOR_CREATION_SUCCEEDED"; break;
      case TRANSFER_SUCCEEDED: temp_msg = "TRANSFER_SUCCEEDED"; break;
      default: temp_msg = "UNKNOWN_ERROR"; break;
   }
   return temp_msg;
}

/** Turn the data store status, block status, process status or error
 * into a human readable string for possible output.
 *
 * @param tstate The state to turn into an error/status message.
 *
 * @return The error/status message. */
const char *DataStore::stateToErrorMessage(DataStore::datastorestate tstate)
{
   const char *temp_msg = "";
   switch (tstate)
   {
      case STORE_CREATED: temp_msg = "this data store has already been created"; break;
      case ALREADY_EXISTS: temp_msg = "this data store already exists"; break;
      case STORE_ACCESS_DENIED: temp_msg = "access to this data store is denied"; break;
      case ERROR_PARSING_SIZES: temp_msg = "error parsing sizes from message"; break;
      case ERROR_PARSING_HASHES: temp_msg = "error parsing hashes from message"; break;
      case ERROR_FINDING_USER: temp_msg = "error finding specified user"; break;
      case NOT_ENOUGH_QUOTA: temp_msg = "no enough quota available for operation"; break;
      case DATA_BLOCK_COMPLETE: temp_msg = "this data block is complete on the server"; break;
      case DATA_BLOCK_INCOMPLETE: temp_msg = "this data block is incomplete on the server"; break;
      case INVALID_PLAINTEXT_HASH_SPECIFIED: temp_msg = "plaintext hash invalid or not understood"; break;
      case WRONG_PLAINTEXT_HASH: temp_msg = "the plaintext hash promised does not match"; break;
      case INVALID_BLOCK_SIZE_SPECIFIED: temp_msg = "the block size specified is invalid"; break;
      case MISMATCHED_BLOCK_SIZE: temp_msg = "the block size of the data does not match the promised size"; break;
      case NO_ALLOCATED_SPACE: temp_msg = "the space to put the data has not been allocated by the server"; break;
      case MESSAGE_CORRUPT: temp_msg = "the message is corrupt and has not been correctly parsed (?)"; break;
      case HASHES_FILE_ERROR: temp_msg = "there is a problem with the list-of-hashes file"; break;
      case SIZES_FILE_ERROR: temp_msg = "there is a problem with the list-of-sizes file"; break;
      case DATA_FILE_ERROR: temp_msg = "there is a problem with a main data block file"; break;
      case BLOCK_ACCESS_DENIED: temp_msg = "access to this block has been denied"; break;
      case TRANSFER_HASH_MISSING: temp_msg = "missing hash for the actual upload data"; break;
      case TRANSFER_HASH_MISMATCH: temp_msg = "hash check failed for the actual upload data"; break;
      case TRANSFER_PAYLOAD_NUMBER_NOT_FOUND: temp_msg = "payload number not found for the upload data"; break;
      case TRANSFER_PAYLOAD_NOT_FOUND: temp_msg = "upload payload not found at specified payload number"; break;
      case DATA_FILE_UNMOVED: temp_msg = "data file unmoved - cannot complete"; break;
      case FILE_CREATION_FAILED: temp_msg = "failed to create the required files"; break;
      case FILE_CREATION_SUCCEEDED: temp_msg = "file creation succeeded"; break;
      case ACTOR_CREATION_SUCCEEDED: temp_msg = "actor creation succeeded"; break;
      case TRANSFER_SUCCEEDED: temp_msg = "the upload has succeeded"; break;
      default: temp_msg = "an unknown error has occurred"; break;
   }
   return temp_msg;
}
